package com.atguigu.gmall.realtime.app.dwm;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.util.MyKafkaUtil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
import org.apache.flink.cep.CEP;
import org.apache.flink.cep.PatternFlatSelectFunction;
import org.apache.flink.cep.PatternFlatTimeoutFunction;
import org.apache.flink.cep.PatternStream;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.conditions.SimpleCondition;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.util.List;
import java.util.Map;

/**
 * @author: xu
 * @desc: 用户跳出明细统计
 * 程  序：mockLog -> Nginx -> Logger.sh -> Kafka(ZK) -> BaseLogApp -> kafka -> UserJumpDetailApp -> Kafka
 */
public class UserJumpDetailApp {
    public static void main(String[] args) throws Exception {
        // TODO 1.基本环境准备
        // 1.1  准备本地测试流环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // 1.2 设置并行度
        env.setParallelism(1);
        // 1.3 设置Checkpoint
        // env.enableCheckpointing(5000, CheckpointingMode.EXACTLY_ONCE);
        // env.getCheckpointConfig().setCheckpointTimeout(60000);
        // env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3,3000L));
        // env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
        // env.setStateBackend(new FsStateBackend("hdfs://node1:8020/gmall/checkpoint/UserJumpDetailApp"))
        // System.setProperty("HADOOP_USER_NAME", "root");

        // TODO 2.从kafka中读取数据
        String sourceTopic = "dwd_page_log";
        String groupId = "user_jump_detail_group";
        // DataStreamSource<String> kafkaStringStream = env.addSource(MyKafkaUtil.getKafkaSource(sourceTopic, groupId));
        DataStream<String> kafkaStringStream = env.fromElements(
                "{\"common\":{\"mid\":\"101\"},\"page\":{\"page_id\":\"home\"},\"ts\":10000} ",
                "{\"common\":{\"mid\":\"102\"},\"page\":{\"page_id\":\"home\"},\"ts\":12000}",

                "{\"common\":{\"mid\":\"102\"},\"page\":{\"page_id\":\"good_list\",\"last_page_id\":" +
                        "\"home\"},\"ts\":15000} ",

                "{\"common\":{\"mid\":\"102\"},\"page\":{\"page_id\":\"good_list\",\"last_page_id\":" +
                        "\"detail\"},\"ts\":30000} "
        );

        // TODO 3.对Kafka的字符串转换为JSON对象，指定事件时间字段，根据日志数据的mid进行分组
        KeyedStream<JSONObject, String> keyedStream = kafkaStringStream
                .map(JSON::parseObject)
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy.<JSONObject>forMonotonousTimestamps().withTimestampAssigner(
                                (jsonObj, recordTimestamp) -> jsonObj.getLong("ts")
                        ))
                .keyBy(
                        jsonObj -> jsonObj.getJSONObject("common").getString("mid")
                );

        /*
          TODO：计算页面跳出明细，需要满足两个条件
           1.是一个首次访问页面，不是从其它页面跳转过来的，last_page_id == null
           2.首次访问页面10秒内，没有对其它页面再次进行访问
         */
        // TODO 4.配置CEP表达式
        Pattern<JSONObject, JSONObject> pattern = Pattern.<JSONObject>begin("first")
                // 1.是一个首次访问页面，不是从其它页面跳转过来的页面
                .where(new SimpleCondition<JSONObject>() {
                           @Override
                           public boolean filter(JSONObject jsonObj) throws Exception {
                               // 获取last_page_id
                               String lastPageId = jsonObj.getJSONObject("page").getString("last_page_id");
                               // 判断上次访问是否为空，为空代表第一次访问
                               return lastPageId == null || lastPageId.length() == 0;
                           }
                       }
                )
                .next("next")
                // 2.首次访问后，在10秒钟内是否访问了其他的页面
                .where(new SimpleCondition<JSONObject>() {
                           @Override
                           public boolean filter(JSONObject jsonObj) throws Exception {
                               // 获取当前页面的id
                               String pageId = jsonObj.getJSONObject("page").getString("page_id");
                               // 判断当前访问的页面id是否为null
                               return pageId != null && pageId.length() > 0;
                           }
                       }
                )
                // 3.时间限制：10秒内没完成就说明用户跳出了
                .within(Time.seconds(10));

        // TODO 5.根据CEP表达式筛选流
        PatternStream<JSONObject> patternStream = CEP.pattern(keyedStream, pattern);

        // TODO 6.从筛选之后的流中，提取数据将超时数据放到侧输出流中
        OutputTag<String> timeoutTag = new OutputTag<>("timeout", BasicTypeInfo.STRING_TYPE_INFO);
        SingleOutputStreamOperator<String> selectStream = patternStream.flatSelect(
                timeoutTag,
                // 处理超时数据
                new PatternFlatTimeoutFunction<JSONObject, String>() {
                    @Override
                    public void timeout(Map<String, List<JSONObject>> pattern, long timeoutTimestamp, Collector<String> out) throws Exception {
                        // 获取所有符合first的json对象
                        List<JSONObject> jsonObjectList = pattern.get("first");
                        // 注意：在timeout方法中的数据都会被参数1中的标签标记
                        for (JSONObject jsonObject : jsonObjectList) {
                            out.collect(jsonObject.toJSONString());
                        }
                    }
                },
                // 处理的没有超时数据
                new PatternFlatSelectFunction<JSONObject, String>() {
                    @Override
                    public void flatSelect(Map<String, List<JSONObject>> pattern, Collector<String> out) throws Exception {
                        // 我们的不需要统计，没有超时的数据，所以这里不需要写什么代码
                    }
                }
        );

        // TODO 7.从侧输出流中获取超时数据
        DataStream<String> userJumpStream = selectStream.getSideOutput(timeoutTag);
        userJumpStream.print(">>>>>");

        // TODO 8.将跳出数据写回到kafka的DWM层
        String sinkTopic = "dwm_user_jump_detail";
        userJumpStream.addSink(MyKafkaUtil.getKafkaSink(sinkTopic));

        env.execute(UserJumpDetailApp.class.getSimpleName());
    }
}
